package com.rabbitmq.example.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.example.dao.MessageMapper;
import com.rabbitmq.example.entity.Message;
import com.rabbitmq.example.entity.MessageExample;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.InitializingBean;

import java.nio.charset.Charset;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;


/**
 * 该RetryMessageRabbitTemplate  只保证   convertAndSend(String exchange, String routingKey, Object message, CorrelationData correlationData)  的可投递性能力，
 * 其余能力，您应该使用 RabbitTemplate
 */

public class RetryMessageRabbitTemplate extends RabbitTemplateAdapter implements InitializingBean {
    private static Logger logger = LoggerFactory.getLogger(RetryMessageRabbitTemplate.class);
    private ScheduledExecutorService scheduledExecutorService;
    private MessageMapper messageMapper;


    private String contentEncoding = "utf-8";

    public RetryMessageRabbitTemplate(RabbitTemplate rabbitTemplate,MessageMapper messageMapper) {
        super(rabbitTemplate);
        this.messageMapper = messageMapper;
    }

    //消息确认
    final public RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback () {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {

            if(ack == true) {
                Message message = new Message();
                message.setId(correlationData.getId());
                message.setTag(true);
                logger.info("update message tag to true ,correlationData: {}",correlationData.getId());
                messageMapper.updateByPrimaryKeySelective(message);
            }
            logger.info("correlationData: {},发送状态: {}, cause: {}" ,correlationData.getId(),ack,cause);
        }
    };

    final public RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback() {
        @Override
        public void returnedMessage(ReturnedMessage returned) {

        }
    };

    public void saveToDB(String id ,Object object, String exchange, String routingKey) throws JsonProcessingException, IllegalAccessException, NoSuchFieldException {

        MessageConverter messageConverter = this.getRabbitTemplate().getMessageConverter();

        org.springframework.amqp.core.Message tmp = messageConverter.toMessage(object, null);

        MessageProperties messageProperties = tmp.getMessageProperties();

        String contentEncoding = messageProperties.getContentEncoding();


        String bodyString = new String(tmp.getBody(), Charset.forName(contentEncoding != null ? contentEncoding : this.contentEncoding));

        ObjectMapper objectMapper = new ObjectMapper();
        String header = objectMapper.writeValueAsString(messageProperties);

        //此处需要获取message 的id

     /*   Class<?> aClass = object.getClass();
        Field id = aClass.getDeclaredField("id");
        id.setAccessible(true);

        Object value = id.get(object);
        String idString = value.toString();*/
        // 发送消息前进行消息的落库
        Message message = new Message(id, exchange, routingKey, bodyString, header, false, 0, new Date());
        messageMapper.insert(message);
        this.getRabbitTemplate().setConfirmCallback(confirmCallback);
        this.getRabbitTemplate().setReturnsCallback(returnsCallback);
    }



    @Override
    public void afterPropertiesSet() {

        //调用父类
        //super.afterPropertiesSet();

        this.scheduledExecutorService = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                return thread;
            }
        });
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                logger.info("定时任务运行");
                try {
                    MessageExample example = new MessageExample();
                    example.createCriteria().andTagEqualTo(false).andRetryLessThan(5);
                    List<Message> messages = messageMapper.selectByExample(example);
                    for (Message message : messages) {
                        //进行消息的从新发送
                        String exchange = message.getExchange();
                        String routingkey = message.getRoutingkey();

                        String messagebody = message.getMessagebody();

                        String header = message.getHeader();
                        ObjectMapper objectMapper = new ObjectMapper();
                        MessageProperties messageProperties = objectMapper.readValue(header, MessageProperties.class);
                        String contentEncoding = messageProperties.getContentEncoding();

                        Charset charset = Charset.forName(contentEncoding != null ? contentEncoding : RetryMessageRabbitTemplate.this.contentEncoding);

                        byte[] bytes = messagebody.getBytes(charset);


                        CorrelationData correlationData = new CorrelationData(message.getId());


                        org.springframework.amqp.core.Message tmp
                                = new org.springframework.amqp.core.Message(bytes, messageProperties);
                        RetryMessageRabbitTemplate.this.getRabbitTemplate().setConfirmCallback(confirmCallback);
                        RetryMessageRabbitTemplate.this.getRabbitTemplate().setReturnsCallback(returnsCallback);
                        //重新发送消息
                        RetryMessageRabbitTemplate.this.convertAndSend(exchange,routingkey,tmp,correlationData);
                        message.setRetry(message.getRetry() + 1);
                        messageMapper.updateByPrimaryKey(message);
                    }
                } catch (JsonProcessingException e) {
                   logger.error("定时任务执行失败",e);
                }
            }
        },0,1, TimeUnit.MINUTES);
    }




    @Override
    public void convertAndSend(String exchange, String routingKey, Object object, CorrelationData correlationData) throws AmqpException {

        String id = correlationData.getId();
        try {
            this.saveToDB(id,object,exchange,routingKey);
        } catch (Exception e) {
            throw new  AmqpException(e);
        }
        super.convertAndSend(exchange, routingKey, object, correlationData);
    }
}
